package edu.nepu.flink.api.transform;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Date 2024/2/28 20:41
 * @Created by chenshuaijun
 */
public class FilterOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<WaterSensor> elementSource = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 11),
                new WaterSensor("s3", 4L, 5)
        );

        // 过滤掉sensor.id不为s1的数据

        SingleOutputStreamOperator<WaterSensor> filterStream = elementSource.filter(element -> "s1".equals(element.getId()));

        filterStream.print();

        env.execute();
    }
}
